package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;

public class Demo02HBaseJavaAPI {
    Connection conn;
    Admin admin; // 对元数据进行操作：create、drop、list、alter、disable、enable
    Table table; // 先通过conn.getTable来初始化，对表的数据进行操作

    @Before
    public void init() throws IOException {
        Configuration conf = new Configuration();
        // 指定HBase地址
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        // 创建HBase连接
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();

    }

    // 列出所有表
    @Test
    public void listTables() throws IOException {
        for (TableName tableName : admin.listTableNames()) {
            System.out.println(tableName.getNameAsString());
        }
    }


    // 删除表
    @Test
    public void dropTable() throws IOException {
        TableName stu = TableName.valueOf("stu");
        // 判断表是否存在
        if (admin.tableExists(stu)) {
            admin.disableTable(stu);
            admin.deleteTable(stu);
            System.out.println("表已删除");
        } else {
            System.out.println("表已存在");
        }
    }

    // 创建表：stu,info
    @Test
    public void createStuTable() throws IOException {
        TableName stu = TableName.valueOf("stu");
        // 判断表是否存在
        if (!admin.tableExists(stu)) {
            // 新版本建表方式
            admin.createTable(
                    TableDescriptorBuilder
                            // 通过表名构建TableDescriptor
                            .newBuilder(stu)
                            // 指定表的列簇信息
                            .setColumnFamily(
                                    ColumnFamilyDescriptorBuilder
                                            .newBuilder("info".getBytes(StandardCharsets.UTF_8))
                                            .setMaxVersions(3)
                                            .setInMemory(true)
                                            .build()
                            ).build()
            );

            // 老版本
//            admin.createTable(new HTableDescriptor(stu)
//                    .addFamily(new HColumnDescriptor("info")
//                            .setMaxVersions(10)
//                    )
//            );
        } else {
            System.out.println("表已存在");
        }
    }

    // 获取表的信息
    @Test
    public void descTable() throws IOException {
        TableDescriptor stu = admin.getDescriptor(TableName.valueOf("stu"));
        // 表可能存在多个列族
        for (ColumnFamilyDescriptor columnFamily : stu.getColumnFamilies()) {
            String cfName = columnFamily.getNameAsString();
            System.out.println(cfName + "列簇的版本号为：" + columnFamily.getMaxVersions());
        }
    }

    // 修改表结构：stu
    // info修改版本为2
    // 再增加一个列簇info2并设置过期时间100s
    @Test
    public void alterStuTable() throws IOException {
        TableName stu = TableName.valueOf("stu");
        TableDescriptor stuDesc = admin.getDescriptor(stu);
        ColumnFamilyDescriptor infoDesc = stuDesc.getColumnFamily("info".getBytes(StandardCharsets.UTF_8));
        ColumnFamilyDescriptor newInfoDesc = ColumnFamilyDescriptorBuilder
                // 需要从原有的Desc基础上进行修改
                .newBuilder(infoDesc)
                .setMaxVersions(4)
                .build();
        admin.modifyColumnFamily(stu, newInfoDesc);

        admin.addColumnFamily(stu, ColumnFamilyDescriptorBuilder
                // 新建一个Desc
                .newBuilder("info2".getBytes(StandardCharsets.UTF_8))
                .setTimeToLive(10)
                .build());

    }

    // 通过Table对象对数据的操作：增删改查
    @Test
    public void put() throws IOException {
        Table stu = conn.getTable(TableName.valueOf("stu"));

        Put put = new Put("002".getBytes(StandardCharsets.UTF_8));
        put.addColumn("info".getBytes(), "id".getBytes(), "001".getBytes());
        put.addColumn("info".getBytes(), "name".getBytes(), "中文".getBytes());
        put.addColumn("info".getBytes(), "age1".getBytes(), "22".getBytes());
        put.addColumn("info".getBytes(), "age2".getBytes(), Bytes.toBytes(22));

        stu.put(put);

    }

    @Test
    public void get() throws IOException {
        Table stu = conn.getTable(TableName.valueOf("stu"));
        Get get1 = new Get("001".getBytes());
        Get get2 = new Get("002".getBytes());
        ArrayList<Get> arr = new ArrayList<>();
        arr.add(get1);
        arr.add(get2);
        Result[] results = stu.get(arr);
        // rs 表示一条数据
        for (Result rs : results) {

            // 假定每条数据结构一致，可以直接通过getRow/getValue提取数据
            String rk = Bytes.toString(rs.getRow());
            String id = Bytes.toString(rs.getValue("info".getBytes(), "id".getBytes()));
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String name1 = Bytes.toString(rs.getValue("info".getBytes(), "name2".getBytes()));
            String age1 = Bytes.toString(rs.getValue("info".getBytes(), "age1".getBytes()));
            int age2 = Bytes.toInt(rs.getValue("info".getBytes(), "age2".getBytes()));

            System.out.println(rk + "," + id + "," + name + "," + name1 + "," + age1 + "," + age2);

            // 如果结构不一致
            // cell表示一个单元格
            for (Cell cell : rs.listCells()) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(cf + "," + qualifier + "," + rowkey + "," + value);
            }
        }
    }

    // Scan
    @Test
    public void scan() throws IOException {
        TableName stu = TableName.valueOf("stu");
        Table stuTable = conn.getTable(stu);

        Scan scan = new Scan();
        // 限制返回的条数
        scan.setLimit(10);
        // 指定RK的范围进行scan
        scan.withStartRow("001".getBytes());
        scan.withStopRow("003".getBytes());
        ResultScanner scanner = stuTable.getScanner(scan);
        for (Result rs : scanner) {
            for (Cell cell : rs.listCells()) {
                String rk = Bytes.toString(CellUtil.copyRow(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rk + "," + value);
            }
        }
    }

    // 需求一：将students.txt数据写入HBase
    // students,info
    @Test
    public void writeStudents() throws IOException {
        TableName stu = TableName.valueOf("students");
        // 建表
        if (!admin.tableExists(stu)) {
            admin.createTable(
                    TableDescriptorBuilder
                            .newBuilder(stu)
                            .setColumnFamily(
                                    ColumnFamilyDescriptorBuilder
                                            .newBuilder("info".getBytes())
                                            .build()
                            ).build());
        } else {
            admin.disableTable(stu);
            admin.deleteTable(stu);
        }

        // 获取Table对象
        Table stuTable = conn.getTable(stu);


        // 写入数据
        // 从文件读取学生数据
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line;
        while ((line = br.readLine()) != null) {
            String[] split = line.split(",");
            // 以学号作为rowkey
            Put put = new Put(split[0].getBytes());
            // 设置每一列
            put.addColumn("info".getBytes(), "name".getBytes(), split[1].getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), split[2].getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), split[3].getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), split[4].getBytes());
            stuTable.put(put);
        }


    }

    // 电信需求：通过手机号查询用户的最新三个位置记录
    @Test
    public void writeDianxin() throws IOException {
        TableName dianxin = TableName.valueOf("dianxin_split");
        // 建表
//        if (!admin.tableExists(dianxin)) {
//            admin.createTable(
//                    TableDescriptorBuilder
//                            .newBuilder(dianxin)
//                            .setColumnFamily(
//                                    ColumnFamilyDescriptorBuilder
//                                            .newBuilder("pos".getBytes())
//                                            .setMaxVersions(3) // 通过多版本来保存多个位置记录
//                                            .build()
//                            ).build());
//        } else {
//            admin.disableTable(dianxin);
//            admin.deleteTable(dianxin);
//        }

        // 获取Table对象
        Table dianxinTable = conn.getTable(dianxin);

        // 写入数据
        // 从文件读取电信数据
        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line;
        // 构建List保存一批put对象
        ArrayList<Put> putList = new ArrayList<>();
        int batchSize = 10000;
        int cnt = 0;
        while ((line = br.readLine()) != null) {
            String[] split = line.split(",");
            String mdn = split[0];
            long startDate = Long.parseLong(split[2]);
            String lg = split[4];
            String la = split[5];

            // 以mdn手机号作为rowkey
            Put put = new Put(mdn.getBytes());
            // 设置每一列
            // 已开始时间作为时间戳
            put.addColumn("cf1".getBytes(), "lg".getBytes(), startDate, lg.getBytes());
            put.addColumn("cf1".getBytes(), "la".getBytes(), startDate, la.getBytes());
            // 将put对象添加到putList中
            putList.add(put);
            if (putList.size() == batchSize) {
                // 提交到HBase
                dianxinTable.put(putList);
                // 清空putList
                putList.clear();
                System.out.println("已经写入了" + cnt + "条数据");
            }
            cnt++;
        }

        // 将剩余的put写入到HBase中
        if (!putList.isEmpty()) {
            dianxinTable.put(putList);
        }
    }

    // 通过手机号mdn查询
    @Test
    public void getPosByMdn() throws IOException {
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        String mdn = "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7";
        Get get = new Get(mdn.getBytes());
//        Get allVersionGet = get.readAllVersions();
        Get newAllVersionGet = get.readVersions(3);

        Result rs = dianxin.get(newAllVersionGet);

//        // 如果直接使用rs.getValue只能取到最新的数据
//        String lg = Bytes.toString(rs.getValue("pos".getBytes(), "lg".getBytes()));
//        String la = Bytes.toString(rs.getValue("pos".getBytes(), "la".getBytes()));
//
//        System.out.println(lg + "," + la);
        ArrayList<String> lgList = new ArrayList<>();
        ArrayList<String> laList = new ArrayList<>();
        for (Cell cell : rs.listCells()) {
            String qua = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            if ("la".equals(qua)) {
                laList.add(value);
            } else {
                lgList.add(value);
            }
        }

        for (int i = 0; i < lgList.size(); i++) {
            String lg = lgList.get(i);
            String la = laList.get(i);
            System.out.println(lg + "," + la);
        }


    }

    @After
    public void close() throws IOException {
        // 关闭连接
        conn.close();
    }
}
